package cn.careerx.app.common.models;

import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.avaje.ebean.Ebean;
import com.avaje.ebean.EbeanServer;
import com.avaje.ebean.EbeanServerFactory;
import com.avaje.ebean.SqlQuery;
import com.avaje.ebean.SqlRow;
import com.avaje.ebean.config.DataSourceConfig;
import com.avaje.ebean.config.ServerConfig;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

import libs.chash.HashAlgorithm;
import libs.chash.KetamaNodeLocator;
import libs.chash.Node;
import libs.chash.NodeImpl;

public class Database {
	
	private final static Logger logger = LoggerFactory.getLogger(Database.class);
	
	private static Lock lock = new ReentrantLock();
	
	private static Config dsConfigs;
	
	private static Config defaultDatasourceConfigs;
	
	private static Map<String, Database> clusterDbs = new HashMap<>();
	
	private static Map<String, Map<String, Object>> clusterEbeanConfigs = new HashMap<>();
	
	//<cluster, <database+tableBase, KetamaNodeLocator<String, String>>
	private static Map<String, Map<String,KetamaNodeLocator<String, String>>> overallShardTables = new HashMap<>();
	
	private Set<String> initailizedDbs = new HashSet<>();
	
	//<dbName+tableName, locator>
	private Map<String, KetamaNodeLocator<String, String>> shardTableLocators = new HashMap<>();
	
	//<dbName, locator>
	private Map<String, KetamaNodeLocator<String, EbeanServer>> shardDbs = new HashMap<>();
	
	private String cluster;
	
	static{
		dsConfigs = ConfigFactory.load("datasource.conf");
		try{
			defaultDatasourceConfigs = dsConfigs.getConfig("defaultDatasourceConfigs");
		}catch(Exception e){
			logger.warn("defaultDatasourceConfigs is not found");
		}
		
		try{
			for(Config dsConf : dsConfigs.getConfigList("dsConfigs")){
				try{
					String cluster = "defaultCluster";
					if(dsConf.hasPath("cluster")){
						cluster = dsConf.getString("cluster");
					}
					
					Map<String, Object> configs = clusterEbeanConfigs.get(cluster);
					if(configs == null){
						configs = new HashMap<>();
						clusterEbeanConfigs.put(cluster, configs);
					}
					
					String name = null;
					
					if(dsConf.hasPath("name")){
						name = dsConf.getString("name").trim();
					}
					
					if(name == null || "".equals(name.trim())){
						logger.error("Incorrect datasource config, name is required.");
						continue;
					}
					
					if(dsConf.hasPath("nodes")){
						
						List<Config> shardConfs = new ArrayList<>();
						for(Config shardConf : dsConf.getConfigList("nodes")){
							
							if(defaultDatasourceConfigs != null){
								shardConf = shardConf.withFallback(defaultDatasourceConfigs);
							}
							
							String shardName = null;
							if(shardConf.hasPath("name")){
								shardName = shardConf.getString("name").trim();
							}
							
							if(shardName == null || "".equals(shardName.trim())){
								logger.error("Incorrect datasource config, name is required for sharded db : " + name);
								continue;
							}
							
							shardConfs.add(shardConf);
						}
						
						configs.put(name, shardConfs);
					}else{
						if(defaultDatasourceConfigs != null){
							dsConf = dsConf.withFallback(defaultDatasourceConfigs);
						}
						configs.put(name, dsConf);
					}
					
					if(dsConf.hasPath("shardTables")){
						//dealing with shard tables.
						try{
							NumberFormat nf = new DecimalFormat("0000");
							Map<String, KetamaNodeLocator<String, String>> locators = overallShardTables.get(cluster);
							if(locators == null){
								locators = new HashMap<>();
							}
							
							boolean needPut = false;
							for(Config conf : dsConf.getConfigList("shardTables")){
								try{
									String shardTableName = conf.getString("name");
									int shardSize = conf.getInt("shardSize");
									if(shardSize < 1){
										throw new Exception("shardSize must > 0.");
									}
									
									List<Node<String, String>> nodes = new ArrayList<>();
									for(int i = 0; i < shardSize; i++){
										String k = shardTableName + nf.format(i+1);
										Node<String, String> node = new NodeImpl<String, String>(k, k);
										nodes.add(node);
									}
									if(nodes.size() == 0){
										continue;
									}
									KetamaNodeLocator<String, String> locator = new KetamaNodeLocator<>(nodes, HashAlgorithm.KETAMA_HASH, 160);
									locators.put(name + "." + shardTableName, locator);
									needPut = true;
								}catch(Exception ex){
									logger.error("error found in shardTables config for dbName = "+name, ex);
								}
							}
							if(needPut && !overallShardTables.containsKey(cluster)){
								overallShardTables.put(cluster, locators);
							}
						}catch(Exception e){
							logger.error("error found in shardTables config for dbName = "+name, e);
						}
					}
				}catch(Exception e){
					logger.error("error parsing datasource config.", e);
				}
			}
		}catch(Exception e){
			logger.error("error parsing datasource configs.", e);
		}
	}
	
	private Database(String cluster) {
		this.cluster = cluster;
		if(overallShardTables.containsKey(cluster)){
			shardTableLocators.putAll(overallShardTables.get(cluster));
		}
	}
	
	public static Database getInstance(String cluster){
		if(cluster == null){
			cluster = "defaultCluster";
		}
		Database database = clusterDbs.get(cluster);
		if(database == null){
			lock.lock();
			try{
				database = clusterDbs.get(cluster);
				if(database == null){
					database = new Database(cluster);
					clusterDbs.put(cluster, database);
				}
			}finally{
				lock.unlock();
			}
		}
		return database;
	}
	
	public EbeanServer getDbByName(String dbName){
		return getDbByPartitionKey(dbName, "");
	}
	
	public EbeanServer getDbByPartitionKey(String dbName, String partitionKey){
		try {
			if(dbName == null){
				return null;
			}
			boolean inited = initailizedDbs.contains(dbName);
			if(!inited){
				EbeanServer ret = null;
				lock.lock();
				try{
					if(!initailizedDbs.contains(dbName)){
						//init from dbConfig
						Map<String, Object> configs = clusterEbeanConfigs.get(cluster);
						if(configs != null){
							Object obj = configs.get(dbName);
							if(obj instanceof Config){
								//build EbeanServer from config
								ret = buildEbeanServerFromConfig((Config)obj);
							}else if(obj instanceof List){
								@SuppressWarnings("unchecked")
								List<Config> shardConfigs = (List<Config>)obj;
								//build locator for db
								List<Node<String, EbeanServer>> nodes = new ArrayList<>();
								for (final Config config : shardConfigs) {
									EbeanServer es = buildEbeanServerFromConfig(config);
									if(es != null){
										nodes.add(new NodeImpl<String, EbeanServer>(config.getString("name"), es));
									}
					            }
								KetamaNodeLocator<String, EbeanServer> locator = new KetamaNodeLocator<>(nodes, HashAlgorithm.KETAMA_HASH, 160);
								Node<String, EbeanServer> node = locator.getPrimary(partitionKey);
								if(node != null){
									ret = node.getObject();
								}
								shardDbs.put(dbName, locator);
							}else{
								logger.warn("unkonw obj:"+obj.getClass().getName());
							}
						}
						initailizedDbs.add(dbName);
					}else{
						KetamaNodeLocator<String, EbeanServer> locator = shardDbs.get(dbName);
						if(locator != null){
							Node<String, EbeanServer> node = locator.getPrimary(partitionKey);
							if(node != null){
								return node.getObject();
							}else{
								return null;
							}
						}else{
							return Ebean.getServer(dbName);
						}
					}
				}catch(Exception e){
					logger.error("DataBase Error",e);
				}finally{
					lock.unlock();
				}
				return ret; 
			}else{
				KetamaNodeLocator<String, EbeanServer> locator = shardDbs.get(dbName);
				if(locator != null){
					Node<String, EbeanServer> node = locator.getPrimary(partitionKey);
					if(node != null){
						return node.getObject();
					}else{
						return null;
					}
				}else{
					return Ebean.getServer(dbName);
				}
			}
		} catch (Exception e) {
			logger.error("DataBase error", e);
		}
		return null;
	}
	
	public Map<EbeanServer, List<String>> getGroupedDbs(String dbName, Iterable<String> partitionKeys){
		Map<EbeanServer, List<String>> ret = new HashMap<>();
		if(partitionKeys != null){
			for(String pKey : partitionKeys){
				EbeanServer es = getDbByPartitionKey(dbName, pKey);
				List<String> pKeys = ret.get(es);
				if(pKeys == null){
					pKeys =  new ArrayList<String>();
					ret.put(es, pKeys);
				}
				pKeys.add(pKey);
			}
		}
		return ret;
	}
	
	public Map<String, List<String>> getGroupedTables(String dbName,String tableBase, Iterable<String> partitionKeys) throws Exception{
		Map<String, List<String>> ret = new HashMap<>();
		if(partitionKeys != null){
			for(String pKey : partitionKeys){
				String table = getTableByPartitionKey(dbName, tableBase, pKey);
				List<String> pKeys = ret.get(table);
				if(pKeys == null){
					pKeys =  new ArrayList<String>();
					ret.put(table, pKeys);
				}
				pKeys.add(pKey);
			}
		}
		return ret;
	}
	
	private static EbeanServer buildEbeanServerFromConfig(Config conf){
		 try {
			final String name = conf.getString("name");
	        final String username = conf.getString("username");
	        final String password = conf.getString("password");
	        final String databaseUrl = conf.getString("databaseUrl") + name + "?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true";
	        final String databaseDriver = conf.getString("databaseDriver");
	        final int minConnections = conf.getInt("minConnections");
	        final int maxConnections = conf.getInt("maxConnections");
	        final String heartbeatsql = conf.getString("heartbeatsql");
	        final boolean defualtServer = conf.getBoolean("defaultServer");
	        final ServerConfig userServerConfig = new ServerConfig();
	        userServerConfig.setName(name);
	
	        final DataSourceConfig userDSConfig = new DataSourceConfig();
	        userDSConfig.setDriver(databaseDriver);
	        userDSConfig.setUsername(username);
	        userDSConfig.setPassword(password);
	        userDSConfig.setUrl(databaseUrl);
	        userDSConfig.setMinConnections(minConnections);
	        userDSConfig.setMaxConnections(maxConnections);
	        userDSConfig.setHeartbeatSql(heartbeatsql);
	        if(conf.hasPath("initConnectionSqls")){
	        	Map<String, String> customProps = new HashMap<>();
	        	customProps.put("initConnectionSqls", conf.getString("initConnectionSqls"));
	            userDSConfig.setCustomProperties(customProps);
	        }
	        userServerConfig.setDataSourceConfig(userDSConfig);
	        userServerConfig.setDefaultServer(defualtServer);
	        userServerConfig.setRegister(true);
            EbeanServer es =  EbeanServerFactory.create(userServerConfig);
            return es;
        } catch (Exception e) {
            logger.error("Error creating EbeanServer from given config: "+conf, e);
            return null;
        }
	}
	
	public String getTableByPartitionKey(String dbName, String tableBase, String partitionKey){
		try {
			String key = dbName + "." + tableBase;
			KetamaNodeLocator<String, String> locator = shardTableLocators.get(key);
			if(locator == null){
				return tableBase;
			}else{
				Node<String,String> n = locator.getPrimary(partitionKey);
				if(n != null){
					return n.getObject();
				}
				return tableBase;
			}
		} catch (Exception e) {
			logger.error("getTableError",e);
		}
		return null;
	}
	
	public static void main(String[] args) {
		Database db = getInstance("Dev");
		EbeanServer es = db.getDbByPartitionKey("CAREERX", "ok");
		SqlQuery query = es.createSqlQuery("select * from CX_USER");
		List<SqlRow> results = query.findList();
		for(SqlRow r : results){
			System.err.println(r.getString("name"));
		}
	}
}
